#include "poolqueue.h"

TaskQueue* QueueInit(int totalsize)
{
    TaskQueue* taskq = (TaskQueue*)malloc(sizeof(TaskQueue));
    if (taskq == NULL)
    {
        return NULL;
    }
    
    if (pthread_mutex_init(&taskq->qmutex, NULL) != 0 || pthread_cond_init(&taskq->qcond, NULL) != 0)
    {
        printf("mutex or condition init fail...\n");
        return NULL;
    }

    taskq->nobloking = false;
    taskq->queueCapacity = totalsize;
    taskq->queueCurSize = 0;
    taskq->queueFront = 0;
    taskq->queueRear = 0;
    taskq->taskQ = (Task*)malloc(sizeof(Task) * totalsize);
    if (taskq->taskQ == NULL)
    {
        free(taskq);
        pthread_mutex_destroy(&taskq->qmutex);
        pthread_cond_destroy(&taskq->qcond);
        return NULL;
    } 
    return taskq;
}

static void DestroyClear(TaskQueue* queue)
{
    pthread_mutex_destroy(&queue->qmutex);
    pthread_cond_destroy(&queue->qcond);
    free(queue->taskQ);
    free(queue);
}

bool QueuePushTask(TaskQueue* queue,Task *task)
{
    if (queue == NULL || task == NULL)
    {
        return false;
    }

    pthread_mutex_lock(&queue->qmutex);
    if (queue->queueCurSize == queue->queueCapacity && queue->nobloking == false)
    {
        pthread_cond_wait(&queue->qcond,&queue->qmutex);
    }
    if (queue->queueCurSize == queue->queueCapacity)
    {
        pthread_mutex_unlock(&queue->qmutex);
        DestroyClear(queue);
        return false;
    }
    
    queue->taskQ[queue->queueRear] = *task;
    queue->queueRear = (queue->queueRear + 1) % queue->queueCapacity;
    queue->queueCurSize++;
    pthread_cond_signal(&queue->qcond);
    pthread_mutex_unlock(&queue->qmutex);
    return true;
}

bool QueuePopTask(TaskQueue* queue,Task *task)
{
    if (queue == NULL || task == NULL)
    {
        return false;
    }

    pthread_mutex_lock(&queue->qmutex);
    if (queue->queueCurSize == 0 && queue->nobloking == false)
    {
        pthread_cond_wait(&queue->qcond,&queue->qmutex);
    }
    if (queue->queueCurSize == 0)
    {
        pthread_mutex_unlock(&queue->qmutex);
        DestroyClear(queue);
        return false;
    }
    *task = queue->taskQ[queue->queueFront];
    queue->queueFront = (queue->queueFront + 1) % queue->queueCapacity;
    queue->queueCurSize--;
    pthread_cond_signal(&queue->qcond);
    pthread_mutex_unlock(&queue->qmutex);
    return true;
}

void QueueCancel(TaskQueue *queue)
{
    queue->nobloking = true;
    pthread_cond_broadcast(&queue->qcond);
}
